package implementation1.app.streams;

import exceptions.StreamEOFException;
import exceptions.StreamException;
import exceptions.StreamReadingException;
import interfaces.app.IStream;

import java.util.ArrayList;
import java.util.ResourceBundle;
import java.util.concurrent.Semaphore;

/**
 * Creates and manages stream which takes output from one process and passes it to another process on input.
 * Created: 24.10.11
 *
 * @author Jan Svab
 */
public class PipeStream implements IStream
{
    /**
     * Resource of messages.
     */
    private ResourceBundle bundle = ResourceBundle.getBundle("implementation1.res.messages");

    /**
     * TRUE - stream is exclusive, FALSE - stream isn't exclusive
     */
    private boolean isExclusive;
    /**
     * Used like FIFO queue. Into this list are saved lines from input. This lines are removed during reading the stream.
     */
    private ArrayList<String> buffer;
    /**
     * Reference to thread which opened stream to writing
     */
    private Thread threadWriter;
    /**
     * Reference to thread which opened stream to reading
     */
    private Thread threadReader;
    /**
     * TRUE - stream is opened to writing
     */
    private boolean openToWrite;
    /**
     * TRUE - stream is opened to reading
     */
    private boolean openToRead;
    /**
     * Semaphore for opening the stream for reading with only one process.
     */
    private Semaphore mutexReader;
    /**
     * Semaphore for opening the stream for writing with only one process.
     */
    private Semaphore mutexWriter;

    /**
     * Default constructor.
     */
    public PipeStream()
    {
        openToWrite = false;
        openToRead = false;
        this.isExclusive = false;

        buffer = new ArrayList<String>();

        threadWriter = null;
        threadReader = null;

        mutexReader = new Semaphore(1);
        mutexWriter = new Semaphore(1);
    }

    /**
     * Return true if the stream is opened.
     *
     * @return true if the stream is opened; otherwise false
     */
    @Override
    public boolean isOpen()
    {
        return openToRead || openToWrite;
    }

    /**
     * Return true if the stream is opened exclusively.
     *
     * @return true if the stream is opened exclusively; otherwise false
     */
    @Override
    public boolean isExclusive()
    {
        return isExclusive;
    }

    /**
     * Open the stream, if no possible return false.
     *
     * @param accessAttribute access to the stream; 1=read; 2=write
     * @return true if opening was successful; otherwise false
     */
    @Override
    public synchronized boolean tryOpen(int accessAttribute)
    {
        if (accessAttribute == AccessAttributes.READ)
        {
            if (mutexReader.availablePermits() == 0)
            {
                return false;
            }
        }

        if (accessAttribute == AccessAttributes.WRITE)
        {
            if (mutexWriter.availablePermits() == 0)
            {
                return false;
            }
        }

        try
        {
            open(accessAttribute);
        } catch (StreamException e)
        {
            return false;
        }

        return true;
    }

    /**
     * Open the stream, if no possible then wait.
     *
     * @param accessAttribute access to the stream; 1=read; 2=write
     * @throws StreamException Thrown if occurs error during opening of the stream.
     */
    @Override
    public void open(int accessAttribute) throws StreamException
    {
        if (accessAttribute != AccessAttributes.READ && accessAttribute != AccessAttributes.WRITE)
        {
            throw new StreamException(bundle.getString("error.stream.openOnlyForReadOrWrite"));
        }

        if (accessAttribute == AccessAttributes.WRITE)
        {
            try
            {
                mutexWriter.acquire();
            } catch (InterruptedException e)
            {
                throw new StreamException(bundle.getString("error.stream.pipeInterruptOpen"));
            }

            threadWriter = Thread.currentThread();
            openToWrite = true;
        }

        if (accessAttribute == AccessAttributes.READ)
        {
            try
            {
                mutexReader.acquire();
            } catch (InterruptedException e)
            {
                throw new StreamException(bundle.getString("error.stream.pipeInterruptOpen"));
            }

            threadReader = Thread.currentThread();
            openToRead = true;
        }
    }

    /**
     * Close the stream.
     *
     * @throws StreamException if stream can not be closed
     */
    @Override
    public synchronized void close() throws StreamException
    {
        if (Thread.currentThread().equals(threadWriter))
        {
            buffer.add(null);
            openToWrite = false;
            threadWriter = null;

            mutexWriter.release();
        }
        else if (Thread.currentThread().equals(threadReader))
        {
            openToRead = false;
            threadReader = null;

            mutexReader.release();
        }
        else
        {
            throw new StreamException(bundle.getString("error.stream.pipeClosingWrongThread"));
        }

        notifyAllProcess();
    }

    /**
     * Read a line from the stream if can not read wait.
     *
     * @return the line from stream
     * @throws StreamException    Thrown if occurs error during reading.
     * @throws StreamEOFException Thrown if occurs EOF on input.
     */
    @Override
    synchronized public String readLine() throws StreamException, StreamEOFException
    {
        if (!openToRead)
        {
            throw new StreamReadingException(bundle.getString("error.stream.noForRead"));
        }
        while (buffer.size() == 0)
        {
            synchronized (this)
            {
                try
                {
                    this.wait();
                } catch (InterruptedException e)
                {
                    if (buffer.size() == 0)
                    {
                        throw new StreamEOFException();
                    }
                }
            }
        }

        String line = buffer.remove(0);
        if (line == null)
        {
            throw new StreamEOFException();
        }
        return line;
    }

    /**
     * Write a line to the stream.
     *
     * @param line a line
     * @throws StreamException Thrown if occurs error during writing into stream.
     */
    @Override
    public void writeLine(String line) throws StreamException
    {
        if (!openToWrite)
        {
            throw new StreamReadingException(bundle.getString("error.stream.noForWrite"));
        }

        buffer.add(line);
        notifyAllProcess();
    }

    /**
     * Notify all waiting process.
     */
    @Override
    public void notifyAllProcess()
    {
        synchronized (this)
        {
            this.notifyAll();
        }
    }

    @Override
    public void setExclusive(boolean exclusive)
    {
        isExclusive = exclusive;
    }
}
